package com.netty.common.config;

import com.alibaba.fastjson.JSON;
import com.netty.common.constant.MqConstants;
import com.netty.common.entity.mq.MqMessage;
import com.netty.common.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @author LiuXiaoYu
 * @date 2021/10/17- 19:51
 */
@Configuration
@Slf4j
public class MqConfig {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisUtil redisUtil;
    /**
     * 声明消息直接模式交换机
     * @return
     */
    @Bean(MqConstants.NETTY_MESSAGE_EXCHANGE)
    public DirectExchange nettyMessageDirectExchange(){
        return new DirectExchange(MqConstants.NETTY_MESSAGE_EXCHANGE);
    }

    /**
     * 声明邮箱队列
     * @return
     */
    @Bean(MqConstants.NETTY_MESSAGE_QUEUE)
    public Queue nettyMessageQueue(){
        Map<String,Object> args=new HashMap<>(2);
        //声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange",MqConstants.NETTY_MESSAGE_DEAD_EXCHANGE);
        //声明当前队列的死信路由 key
        args.put("x-dead-letter-routing-key",MqConstants.NETTY_MESSAGE_DEAD_ROUTING_KEY);
        return QueueBuilder.durable(MqConstants.NETTY_MESSAGE_QUEUE).withArguments(args).build();
    }

    /**
     * 邮箱交换机通过邮箱路由绑定到邮箱队列
     * @return 绑定关系
     */
    @Bean
    public Binding bindingNettyMessageQueueToMailExchange(@Qualifier(MqConstants.NETTY_MESSAGE_QUEUE)Queue nettyMessageQueue,
                                                  @Qualifier(MqConstants.NETTY_MESSAGE_EXCHANGE)DirectExchange nettyMessageDirectExchange){
        return BindingBuilder.bind(nettyMessageQueue).to(nettyMessageDirectExchange).with(MqConstants.NETTY_MESSAGE_ROUTING_KEY);
    }

    /**
     * 死信交换机
     * @return
     */
    @Bean(MqConstants.NETTY_MESSAGE_DEAD_EXCHANGE)
    public DirectExchange nettyMessageDeadExchange() {
        return new DirectExchange(MqConstants.NETTY_MESSAGE_DEAD_EXCHANGE);
    }
    /**
     * 声明死信队列
     * @return
     */
    @Bean(MqConstants.NETTY_MESSAGE_DEAD_QUEUE)
    public Queue nettyMessageDeadQueue() {
        return new Queue(MqConstants.NETTY_MESSAGE_DEAD_QUEUE);
    }

    /**
     * 声明死信交换机 死信路由 死信队列 绑定关系
     * @return 绑定关系
     */
    @Bean
    public Binding deadMessageBindingQd(@Qualifier(MqConstants.NETTY_MESSAGE_DEAD_QUEUE) Queue nettyMessageDeadQueue,
                                        @Qualifier(MqConstants.NETTY_MESSAGE_DEAD_EXCHANGE) DirectExchange nettyMessageDeadExchange) {
        return BindingBuilder.bind(nettyMessageDeadQueue).to(nettyMessageDeadExchange).with(MqConstants.NETTY_MESSAGE_DEAD_ROUTING_KEY);
    }

    /**
     * 使用json序列化机制,进行消息转换
     *
     * @return json
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @PostConstruct
    public void initRabbitTemplate() {
        /** 生产者发送消息到交换机 不管成功与否都触发回调
         * correlationData 当前消息的唯一id
         * ack 消息是否成功收到
         * cause 失败的原因
         */
        //设置开启mandatory,才能触发回调函数
        rabbitTemplate.setMandatory(true);
        /**
         * 设置消息确认回调
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            assert correlationData != null;
            MqMessage redisMqMessage = (MqMessage)redisUtil.get(MqConstants.NETTY_MESSAGE_LOG_KEY+correlationData.getId());
            if (redisMqMessage==null){
                //说明日志丢失了
                log.error("存储在redis中的mq日志丢失,消息id:"+correlationData.getId());
                return;
            }
            if (ack){
                log.info("消息成功发送到Exchange，相关数据：{}，原因：{}", JSON.toJSONString(correlationData),cause);
                 redisMqMessage.setStatus(MqConstants.MESSAGE_SUCCESS_TO_EXCHANGE).setMessageId(correlationData.getId());
                //新增消息状态为投递成功
            }else {
                log.error("消息发送到Exchange失败，相关数据：{}，原因：{}", JSON.toJSONString(correlationData),cause);
                //新增消息状态为投递失败
                redisMqMessage.setStatus(MqConstants.MESSAGE_FAIL_TO_EXCHANGE_OR_QUEUE).setMessageId(correlationData.getId());
            }
            //更新redis状态
            redisUtil.set(MqConstants.NETTY_MESSAGE_LOG_KEY+correlationData.getId(),redisMqMessage);
        });

        /** 投递失败回调
         * 只要消息交换机没有投递给指定的队列,就会触发失败回调
         * message 投递失败的消息详细信息
         * replyCode 回复的状态码
         * replyText 回复的文本内容
         * exchange 当时这个消息发给那个交换机
         * routingKey 当时指定的路由键
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.error("消息路由到队列失败,当前消息：{}，回应码：{}，回应信息：{}，交换机：{}，路由Key：{}",
                    JSON.toJSONString(message),replyCode,replyText,exchange,routingKey);
            MqMessage redisMqMessage = (MqMessage)redisUtil.get(MqConstants.NETTY_MESSAGE_LOG_KEY+message.getMessageProperties().getCorrelationId());
            if (redisMqMessage==null) {
                //说明日志丢失了
                log.error("存储在redis中的mq日志丢失,消息id:" + message.getMessageProperties().getCorrelationId());
                return;
            }
           redisMqMessage.setMessageId(message.getMessageProperties().getCorrelationId())
                    .setStatus(MqConstants.MESSAGE_FAIL_TO_EXCHANGE_OR_QUEUE);
            //更新存储在redis中的消息状态
            redisUtil.set(MqConstants.NETTY_MESSAGE_LOG_KEY+message.getMessageProperties().getCorrelationId(),redisMqMessage);
        });
    }

}
